package mr.scan;

import java.io.BufferedReader;
import java.io.FileReader;
import java.io.IOException;
import java.util.Map.Entry;
import java.util.StringTokenizer;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.filecache.DistributedCache;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import util.ScanList;

public class SecondSeqMapper extends
		Mapper<LongWritable, Text, LongWritable, LongWritable> {

	// private ArrayList<Integer> scanList;
	// private Hashtable<Integer, Integer> keyIndex;
	private static final Log LOG = LogFactory.getLog(SecondSeqMapper.class);

	int initFlg = -1;
	private ScanList<Long> psums;
	private long chunkey = -1;

	private long totalsum = 0;

	@Override
	public void map(LongWritable key, Text list, Context context)
			throws IOException, InterruptedException {
		// if (chunkey <0) {
		// chunkey = key.get();
		// Entry<Long, Long> de = psums.floorEntry(chunkey);
		// if(de != null && de.getKey() != chunkey)
		// totalsum += de.getValue();
		// }

		//StringBuffer newList = new StringBuffer(1024);
		StringTokenizer itr = new StringTokenizer(list.toString());
		while (itr.hasMoreTokens()) {
			totalsum += Integer.parseInt(itr.nextToken());
			//newList.append(totalsum + " ");
			context.write(new LongWritable(chunkey), new LongWritable(totalsum));
		}

		// LOG.debug("finish map scanList. Time =" +
		// (System.currentTimeMillis()-start));
		
	}

	@Override
	protected void cleanup(Context context) throws IOException,
			InterruptedException {
		LOG.info("finish a map task ( chunk-id" + chunkey + "), total sum= "
				+ totalsum);
	}

	@Override
	protected void setup(Context context) throws IOException,
			InterruptedException {
		getChunkey(context);

		Path[] paths = DistributedCache.getLocalCacheFiles(context
				.getConfiguration()); // context.getLocalCacheFiles();
		// File f = new File(paths[0].toString());
		// Files.readAllLines(new
		// java.nio.file.Path(paths[0].toString()),StandardCharsets.UTF_8);
		// Charset charset = StandardCharsets.UTF_8;
		String partialSumList = "";
		long start = System.currentTimeMillis();
		BufferedReader reader = new BufferedReader(new FileReader(
				paths[0].toString())); // Files.newBufferedReader(f.toPath(),charset);
		try {

			String line = null;
			while ((line = reader.readLine()) != null) {
				partialSumList += line;
			}
			partialSumList.trim();
			// System.out.println("**************"+partialSumList);
			// take care, maybe very large
			// LOG.debug("**************"+partialSumList);
		} catch (IOException x) {
			System.err.format("IOException: %s%n", x);
		}

		LOG.debug("finish reading partialSumList. Time ="
				+ (System.currentTimeMillis() - start));

		start = System.currentTimeMillis();
		if (partialSumList.length() > 0) {
			String[] tmp = partialSumList.split(";");
			psums = new ScanList<Long>();
			long sum = 0;
			for (String val : tmp) {
				if (val != null && val.length() > 0) {
					long key = Long.parseLong(val.split("=")[0]);
					long parSum = Integer.parseInt(val.split("=")[1]);
					sum += parSum;
					psums.put(key, sum); // scan
				}
			}
		}
		LOG.debug("finish parsing partialSumList. Time ="
				+ (System.currentTimeMillis() - start));

		Entry<Long, Long> de = psums.lowerEntry(chunkey);
		if (de != null)
			totalsum += de.getValue();

	}

	public long getChunkey(Context reporter) {
		String spl = reporter.getInputSplit().toString();
		if (spl != null && spl.length() > 0) {
			String[] offset = spl.substring(spl.lastIndexOf(":")+1).split("\\+");
			chunkey = Long.parseLong(offset[0]);
			// nextChunkey = Long.parseLong(offset[1]);
		}
		return chunkey;
	}
}